package net.zbus.rpc;

import java.lang.reflect.Constructor;
import java.lang.reflect.InvocationHandler;
import java.lang.reflect.Method;
import java.lang.reflect.Proxy;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

import net.zbus.ClientPool;
import net.zbus.ClientPoolConfig;

public class RpcFactory {
	
	private static Constructor<InvocationHandler> rpcProxy;
	private static Map<String, InvocationHandler> cache = new ConcurrentHashMap<String, InvocationHandler>();
	private static Map<String, ClientPool> clientPools = new ConcurrentHashMap<String, ClientPool>();
	private static Map<String, Boolean> ownPools = new ConcurrentHashMap<String, Boolean>();
	
	
	static {
		try {
			Class<?> proxyClass = Class.forName("net.zbus.rpc.RpcProxy");
			@SuppressWarnings("unchecked")
			Class<InvocationHandler> clazz = (Class<InvocationHandler>) proxyClass;
			rpcProxy = clazz.getConstructor(new Class[] {Map.class, ClientPool.class });
		} catch (Exception e) {
			e.printStackTrace();
		}
	}
	
	public static void setClientPool(String broker, ClientPool pool){
		if(clientPools.containsKey(broker)) return;
		clientPools.put(broker, pool); 
	}

	public static Map<String, String> parseKeyValues(String kvstring) {
		Map<String, String> res = new HashMap<String, String>();
		String[] parts = kvstring.split("\\&");
		for(String kv : parts){
			String[] kvp = kv.split("=");
			String key = kvp[0].trim();
			String val = "";
			if(kvp.length>1){
				val = kvp[1].trim();
			} 
			res.put(key, val);
		}
		return res;
	} 
	
	/**
	 * zbus://10.60.60.13:15555/TRADE?encoding=utf8
	 *
	 * parameters after ? got default values
	 * 
	 * encoding=UTF8
	 * module=
	 * timeout=10000
	 * token=
	 * log=true
	 * 
	 * @param api
	 * @param serviceURL
	 */
	@SuppressWarnings("unchecked")
	public static <T> T getService(Class<T> api, String serviceUrl)
			throws Exception {
		String url = serviceUrl.trim();
		final String prefix = "zbus://";
		if (url.startsWith(prefix)) {
			url = serviceUrl.substring(prefix.length());
		}

		ClientPoolConfig config = new ClientPoolConfig();
		String[] parts = url.split("/");
		String broker = parts[0].trim();
		String service = parts[1].trim();

		config.setBroker(broker);
		parts = service.split("\\?");
		
		String serviceName = parts[0].trim();

		Map<String, String> kvs = new HashMap<String, String>();
		if (parts.length > 1) {
			String kvstring = parts[1].trim();
			kvs = parseKeyValues(kvstring);  
		}
		kvs.put("service", serviceName);
		String module = kvs.get("module");
		if( module == null){//use default moudle
			module = api.getSimpleName();
			kvs.put("module", module); 
		}
		
		InvocationHandler handler = cache.get(serviceUrl+module);
		Class<T>[] interfaces = new Class[] { api };
		if(handler != null){
			return (T) Proxy.newProxyInstance(Thread.currentThread()
					.getContextClassLoader(), interfaces, handler);
		}
		
		
		
		ClientPool pool = clientPools.get(broker);
		if(pool == null){
			pool = new ClientPool(config);
			clientPools.put(broker, pool);
			ownPools.put(broker, true);
		}
		
		handler = rpcProxy.newInstance(kvs, pool);
		cache.put(serviceUrl, handler);

		return (T) Proxy.newProxyInstance(Thread.currentThread()
				.getContextClassLoader(), interfaces, handler);
		
	}
	
	public static void destroy(){ 
		for(Map.Entry<String, ClientPool> entry : clientPools.entrySet()){
			String broker = entry.getKey();
			if(ownPools.containsKey(broker) && ownPools.get(broker)){ 
				ClientPool pool = entry.getValue();
				pool.destroy(); 
			}
		}
	}
}

class RpcProxy implements InvocationHandler {
	private String serviceUrl;
	private String serviceName;
	private String encoding = "UTF8";
	private String accessToken = "";
	private String module = "";
	private int timeout = 10000; 
	private boolean zbusLog = true;
	 
	private ClientPool clientPool; 
	private Rpc rpc;
	
	private static final Object REMOTE_METHOD_CALL = new Object();

	public RpcProxy(Map<String, String> kvs, ClientPool clientPool) {
		this.clientPool = clientPool;
		if(kvs.containsKey("service")){
			this.serviceName = kvs.get("service");
		} else {
			throw new RuntimeException("serviceName required");
		}
		
		if (kvs.containsKey("encoding")) {
			this.encoding = kvs.get("encoding");
		}
		if (kvs.containsKey("timeout")) {
			this.timeout = Integer.valueOf(kvs.get("timeout"));
		}
		if (kvs.containsKey("token")) {
			this.accessToken = kvs.get("token");
		}
		if (kvs.containsKey("module")) {
			this.module = kvs.get("module");
		}
		if (kvs.containsKey("log")) {
			this.zbusLog = Boolean.valueOf(kvs.get("log"));
		}
		this.rpc = new Rpc(this.clientPool, this.serviceName);
		this.rpc.setEncoding(this.encoding);
		this.rpc.setTimeout(this.timeout);
		this.rpc.setToken(this.accessToken);
		this.rpc.setModule(this.module);  
		this.rpc.setZbusLog(this.zbusLog); 
	}

	@Override
	public Object invoke(Object proxy, Method method, Object[] args)
			throws Throwable {
		Object value = handleLocalMethod(proxy, method, args);
		if (value != REMOTE_METHOD_CALL) return value;
		return rpc.invoke(method.getName(), args);
	}

	protected Object handleLocalMethod(Object proxy, Method method,
			Object[] args) throws Throwable {
		String methodName = method.getName();
		Class<?>[] params = method.getParameterTypes();

		if (methodName.equals("equals") && params.length == 1
				&& params[0].equals(Object.class)) {
			Object value0 = args[0];
			if (value0 == null || !Proxy.isProxyClass(value0.getClass()))
				return new Boolean(false);
			RpcProxy handler = (RpcProxy) Proxy.getInvocationHandler(value0);
			return new Boolean(this.serviceUrl.equals(handler.serviceUrl));
		} else if (methodName.equals("hashCode") && params.length == 0) {
			return new Integer(this.serviceUrl.hashCode());
		} else if (methodName.equals("toString") && params.length == 0) {
			return "ZBusProxy[" + this.serviceUrl + "]";
		}
		return REMOTE_METHOD_CALL;
	} 
}